package com.hc.programming.thread;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 现在有大量文本文件（100+），预计一共涉及单词数量1W左右。需要由5个线程并发计算，全部计算后做结果合并，选出出现频率最高的100个单词及对应次数。
 *
 * @author huangchao E-mail:18008108748@163.com
 * @version 创建时间：2024/9/5 14:35
 */
public class 多线程多文件单词数量统计 {
    public static void main(String[] args) throws InterruptedException {
        List<String> paths = new ArrayList<>();
        for (int i = 1; i < 16; i++) {
            paths.add("D:\\tmp\\words\\" + i + ".txt");
        }
        System.out.println(new 多线程多文件单词数量统计().wordCount(paths));
    }

    /**
     * 统计前100个单词数量
     *
     * @param path 文件路径列表
     * @return 前100个单词及次数
     */
    public Map<String, Integer> wordCount(List<String> path) throws InterruptedException {
        if (path == null || path.isEmpty()) {
            return new HashMap<>();
        }
        // 线程数量: 最大5个
        int threadCount = Math.min(5, path.size());
        int keepWordCount = 100; // 保留单词数量
        CountDownLatch cdl = new CountDownLatch(threadCount);
        // 1、待合并数组
        Map<String, Integer>[] mergeMapArr = new Map[threadCount];
        // 2、多线程并行处理
        for (int i = 0; i < threadCount; i++) {
            int finalI = i;
            new Thread(() -> {
                // 2.1 读取 & 计算 path[i]、path[i+5]、path[i+10]...
                Map<String, Integer> map = new HashMap<>();
                for (int j = finalI; j < path.size(); j += threadCount) {
                    try (
                            BufferedReader br = new BufferedReader(new InputStreamReader(Files.newInputStream(Paths.get(path.get(j)))))
                    ) {
                        String line;
                        while ((line = br.readLine()) != null) {
                            if (Objects.equals(line.trim(), "")) { // 空行过滤
                                continue;
                            }
                            // 2.1.1 假定以空格分隔
                            for (String word : line.split(" ")) {
                                Integer num = map.getOrDefault(word, 0);
                                map.put(word, num + 1);
                            }
                        }
                    } catch (Exception e) {
                        System.out.println("读取文件报错：" + e.getMessage());
                        e.printStackTrace();
                    }
                }
                // 2.2 map排序，截取前100个 & 返回主程序
                mergeMapArr[finalI] = map.size() > keepWordCount ? sortAndCropping(map, keepWordCount) : map;
                cdl.countDown();
            }).start();
        }
        cdl.await(1, TimeUnit.HOURS); // 最大阻塞1小时
        // 3、合并结果：mergeArr
        Map<String, Integer> rst = mergeMapArr[0];
        for (int i = 1; i < threadCount; i++) {
            for (String word : mergeMapArr[i].keySet()) {
                Integer num = rst.getOrDefault(word, 0);
                rst.put(word, num + mergeMapArr[i].get(word));
            }
        }
        // 4、结果排序，裁剪前100个
        return sortAndCropping(rst, keepWordCount);
    }

    /**
     * 排序 & 裁剪
     *
     * @param map   待操作map
     * @param count 裁剪数量
     * @return 结果
     */
    private Map<String, Integer> sortAndCropping(Map<String, Integer> map, int count) {
        return map.entrySet().stream()
                .sorted(Map.Entry.comparingByValue(Comparator.reverseOrder())) // 降序排序
                .limit(count)
                .collect(Collectors.toMap(
                        Map.Entry::getKey,
                        Map.Entry::getValue,
                        (oldValue, newValue) -> oldValue,
                        LinkedHashMap::new
                ));
    }
}
